package top.tenebrous.termbor.rpc.consumer.handler;

import cn.hutool.json.JSONUtil;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.Getter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import top.tenebrous.termbor.rpc.consumer.context.RpcContext;
import top.tenebrous.termbor.rpc.protocol.enums.RpcType;
import top.tenebrous.termbor.rpc.proxy.future.RPCFuture;
import top.tenebrous.termbor.rpc.protocol.RpcProtocol;
import top.tenebrous.termbor.rpc.protocol.header.RpcHeader;
import top.tenebrous.termbor.rpc.protocol.message.RpcRequest;
import top.tenebrous.termbor.rpc.protocol.message.RpcResponse;
import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @program: termbor-rpc
 * @description:
 * @author: yazhang6
 * @create: 2024-05-22 11:05
 **/
public class RpcConsumerHandler extends SimpleChannelInboundHandler<RpcProtocol<RpcResponse>> {

    private final Logger logger = LoggerFactory.getLogger(RpcConsumerHandler.class);

    @Getter
    private volatile Channel channel;

    @Getter
    private SocketAddress remotePeer;

    private Map<Long, RpcProtocol<RpcResponse>> pendingResponse = new ConcurrentHashMap<>();

    private Map<Long, RPCFuture> pendingRPC = new ConcurrentHashMap<>();

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        this.remotePeer = this.channel.remoteAddress();
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        this.channel = ctx.channel();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext,
        RpcProtocol<RpcResponse> protocol) throws Exception {
        if (protocol == null) {
            return;
        }
        this.handlerMessage(protocol);
        logger.info("服务消费者接收到的数据 ===> {}", JSONUtil.toJsonStr(protocol));

    }

    private void handlerMessage(RpcProtocol<RpcResponse> protocol) {
        RpcHeader header = protocol.getHeader();
        // 心跳消息
        if (header.getMsgType() == (byte)RpcType.HEARTBEAT_TO_CONSUMER.getType()) {
            this.handlerHeartbeatMessage(protocol);
        } else if (header.getMsgType() == (byte) RpcType.RESPONSE.getType()) {
            // 响应消息
            this.handlerResponseMessage(protocol, header);
        }
    }

    private void handlerResponseMessage(RpcProtocol<RpcResponse> protocol, RpcHeader header) {
        long requestId = header.getRequestId();
        RPCFuture rpcFuture = pendingRPC.remove(requestId);
        if (rpcFuture != null) {
            rpcFuture.done(protocol);
        }
    }

    private void handlerHeartbeatMessage(RpcProtocol<RpcResponse> protocol) {
        // 心跳消息处理
        logger.info("receive service provider hearbeat message:{}", protocol.getBody().getResult());
    }

    public RPCFuture sendRequest(RpcProtocol<RpcRequest> protocol, boolean async, boolean oneway) {
        logger.info("服务消费者发送的数据 ===> {}", JSONUtil.toJsonStr(protocol));
        return oneway ? sendRequestOneway(protocol) : (async ? sendRequestAsync(protocol) : sendRequestSync(protocol));
    }

    private RPCFuture getRPpcFuture(RpcProtocol<RpcRequest> protocol) {
        RPCFuture rpcFuture = new RPCFuture(protocol);
        RpcHeader header = protocol.getHeader();
        long requestId = header.getRequestId();
        pendingRPC.put(requestId, rpcFuture);
        return rpcFuture;
    }

    private RPCFuture sendRequestSync(RpcProtocol<RpcRequest> protocol) {
        RPCFuture rPpcFuture = this.getRPpcFuture(protocol);
        channel.writeAndFlush(protocol);
        return rPpcFuture;
    }

    private RPCFuture sendRequestAsync(RpcProtocol<RpcRequest> protocol) {
        RPCFuture rPpcFuture = this.getRPpcFuture(protocol);
        // 如果是异步调用，则将 RPCFuture 放入 RpcContext
        RpcContext.getContext().setRPCFuture(rPpcFuture);
        channel.writeAndFlush(protocol);
        return null;
    }

    private RPCFuture sendRequestOneway(RpcProtocol<RpcRequest> protocol) {
        channel.writeAndFlush(protocol);
        return null;
    }

    public void close() {
        channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }
}
